package com.edgewalk.rabbitmq.basic.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalTime;

public class RpcServer {
	private static final String RPC_QUEUE_NAME = "rpc_queue";

	public static void execute(String host, String userName, String password) {
		// 配置连接工厂
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(host);
		// 使用管理后台默认guest帐号
		factory.setUsername(userName);
		factory.setPassword(password);
		try (
				// 建立TCP连接
				Connection connection = factory.newConnection();
				// 在TCP连接的基础上创建通道
				final Channel channel = connection.createChannel();
		) {
			// 声明一个rpc_queue队列
			channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
			// 设置同时最多只能获取一个消息
			channel.basicQos(1);
			System.out.println(" [RpcServer] Awaiting RPC requests");

			// 定义消息的回调处理类
			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
					// 生成返回的结果，关键是设置correlationId值
					AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
							.correlationId(properties.getCorrelationId())
							.build();
					System.out.println("["+LocalDate.now()+" "+LocalTime.now() +"] [RpcServer]-["+RPC_QUEUE_NAME+"] 收到消息: " + new String(body)+",correlationId="+properties.getCorrelationId());
					try {
						//执行处理逻辑
						Thread.sleep(1000*3);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					// 生成返回
					String response = "corrleatinId="+properties.getCorrelationId()+"的消息处理完成";
					// 回复消息，通知已经收到请求
					channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
					System.out.println("["+LocalDate.now()+" "+LocalTime.now() +"] [RpcServer]-["+properties.getReplyTo()+"] 发送消息: " + response);
					// 对消息进行应答
					channel.basicAck(envelope.getDeliveryTag(), false);
					// 唤醒正在消费者所有的线程
					synchronized (this) {
						this.notify();
					}
				}
			};
			// 关闭自动确认收到消息,采用手动确认channel.basicAck
			channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

			// 在收到消息前，本线程进入等待状态
			while (true) {
				synchronized (consumer) {
					try {
						consumer.wait();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 暂停3s，并返回结果
	 *
	 * @param body
	 * @return
	 */
	private static String generateResponse(byte[] body) {
		try {
			Thread.sleep(1000 * 3);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return "response:" + new String(body) + "-" + System.currentTimeMillis();
	}
}
